package com.atguigu.flink.wordcount;

import com.atguigu.flink.pojo.WordCount;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/1/12
 *
 *
 *  @FunctionalInterface 都支持使用lamda表达式创建实例！
 *
 *          非法类型异常
 *          Exception in thread "main" org.apache.flink.api.common.functions.InvalidTypesException:
 *                  flatMap 返回的类型由于泛型擦除无法自动确认。
 *                      原因是因为 转换之前 Collector<WordCount>，之后使用lamda表达式，由于泛型擦除。
 *                              Collector<WordCount> ----> Collector<Object>
 *                              由Collector<Object> 无法确定 ，向下游发送的是 WordCount
 *
 *              The return type of function 'main(Demo6_LamdaPOJO.java:28)'
 *              could not be determined automatically,due to type erasure.
 *
 *              解决方案：  在转换得到结果后，调用 returns()，明确声明flatmap返回的泛型类型！
 *
 *              You can give type information hints by using the returns(...) method on the result of the transformation call,
 *              or by letting your function implement the 'ResultTypeQueryable' interface.
 *
 */
public class Demo6_LamdaPOJO
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        System.out.println("hello ............");

        env.setParallelism(1);

        DataStreamSource<String> ds = env.socketTextStream("hadoop103",8888);

        SingleOutputStreamOperator<WordCount> ds1 = ds
            .flatMap( (String line, Collector<WordCount> out) -> {
                String[] words = line.split(" ");
                for (String word : words) {
                        out.collect(new WordCount(word,1));
                    }
                })
            //类型信息提示
            .returns(WordCount.class).setParallelism(10);


        /*
            调用WordCount的什么方法，返回一个key

            方法引用是lamda的一种特殊情况的实现。
                当函数式接口的抽象方法的实现和某个现成的方法的实现逻辑一致！可以直接引用现成的某个方法！
                    格式:  类名::方法名()
         */
        ds1.keyBy(WordCount::getWord)
          .sum("count")
          .print();


        env.execute();


    }
}
